package datax.hook;

import com.alibaba.fastjson.JSON;
import com.qihoo.finance.govern.common.database.constants.DbType;
import com.qihoo.finance.govern.modules.metadata.enums.JobType;
import com.qihoo.finance.govern.modules.metadata.enums.SyncLogEventType;
import com.qihoo.finance.govern.modules.metadata.mq.SyncLogMsg;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Properties;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;

/**
 * @author: luozehuan
 * @date: 2022/06/07
 **/
public class SyncLogProducer {

    private final Logger logger = LoggerFactory.getLogger(this.getClass());

    private static Properties properties = new Properties();

    static {
        properties.put("key.serializer",
                "org.apache.kafka.common.serialization.StringSerializer");
        properties.put("value.serializer",
                "org.apache.kafka.common.serialization.StringSerializer");
        properties.put("bootstrap.servers", "localhost:9092");
        properties.put("retries", 1024);
    }

    public void send(String jsonMsg,String topic){
        try {
        KafkaProducer producer = initProducer(properties);
        ProducerRecord<String, String> record = new ProducerRecord<>(topic, jsonMsg);
        Future send = producer.send(record);
            send.get();
        } catch (Exception e) {
            logger.info("Hook: send msg error:{}",e);
        }
    }

    private KafkaProducer initProducer(Properties properties){
        return new KafkaProducer<>(properties);
    }
}
